package com.cll.learn.rpc.consumer.context;

import com.cll.learn.rpc.common.coder.RpcDecoder;
import com.cll.learn.rpc.common.coder.RpcRequestEncoder;
import com.cll.learn.rpc.common.dto.RpcRequest;
import com.cll.learn.rpc.common.dto.RpcResponse;
import com.cll.learn.rpc.common.serializer.JsonSerializable;
import com.cll.learn.rpc.common.serializer.ProtostuffSerializable;
import com.cll.learn.rpc.consumer.handler.RpcConsumerHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 描述信息:
 * 客户端连接服务端
 * 客户端向服务端提交请求
 * @author CLL
 * @version 1.0
 * @date 2020/9/14 15:51
 */
@RequiredArgsConstructor
@Getter
public class RpcConsumer {

    private static final Logger logger = LoggerFactory.getLogger(RpcConsumer.class);

    @NonNull
    private String ip;
    @NonNull
    private Integer port;

    /**
     * 提交任务的线程池
     */
    private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors()
            , (int) (Runtime.getRuntime().availableProcessors() * 1.2)
            , 5000
            , TimeUnit.MILLISECONDS
            , new ArrayBlockingQueue<>(100),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.DiscardPolicy()
    );

    /**
     * 客户端发送接收请求的netty通道处理器
     */
    private static RpcConsumerHandler rpcConsumerHandler;

    @SuppressWarnings("unchecked")
    public <T> T createProxy(final Class<?> serviceInterface){
        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader()
                , new Class[]{serviceInterface}
                , new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        logger.info("===>>> proxy object invoke");
                        if (null == RpcConsumer.rpcConsumerHandler) {
                            initClient(getIp(), getPort());
                        }

                        RpcRequest rpcRequest = new RpcRequest(UUID.randomUUID().toString()
                                , method.getDeclaringClass().getName()
                                , method.getName()
                                , method.getParameterTypes()
                                , args);

                        if (null == RpcConsumer.rpcConsumerHandler) {
                            logger.error("****** 客户端异步启动还没有完成");
                        }
                        RpcConsumer.rpcConsumerHandler.setParam(rpcRequest);
                        Class<?> returnType = method.getReturnType();
                        Object result = threadPool.submit(RpcConsumer.rpcConsumerHandler).get();
                        if (null != result) {
                            if (returnType.isInstance(result)) {
                                logger.info("===>>> type success");
                            } else {
                                logger.error("****** type error!");
                            }
                        }
                        return result;
                    }
                });
    }

    /**
     * 初始化netty客户端
     */
    public static void initClient(String ip, int port) throws InterruptedException {
        RpcConsumer.rpcConsumerHandler = new RpcConsumerHandler();

        NioEventLoopGroup clientCroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(clientCroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 编码客户端发送给服务提供者的RpcRequest对象
                        pipeline.addLast(new RpcRequestEncoder(RpcRequest.class, new ProtostuffSerializable()));
                        // 解码服务提供者返回的RpcResponse对象
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new ProtostuffSerializable()));

                        pipeline.addLast(RpcConsumer.rpcConsumerHandler);
                    }
                });
        bootstrap.connect(ip, port).sync();
    }
}